package com.li.spark

import org.apache.spark.{SparkConf, SparkContext}

object ActionOp {
  def main(args: Array[String]): Unit = {
    val sc: SparkContext = getSparkContext
    //    reduceOp(sc)
    //    collectOp(sc)
    //    takeOp(sc)
    //    countOp(sc)
    //    saveAsTextFileOp(sc)
    //    countByKeyOp(sc)
    sc.stop();
  }

  def countByKeyOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array((1, 2), (1, 3), (2, 3)));
    dataRDD.countByKey().foreach(println(_))

  }

  def saveAsTextFileOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1, 2, 3, 4, 5));
    dataRDD.saveAsTextFile("/Users/lijiacen/Downloads/example.txt")
  }

  def countOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1, 2, 3, 4, 5));
    val count = dataRDD.count();
    println(count)
  }

  def takeOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1, 2, 3, 4, 5));
    val array = dataRDD.take(3);
    array.foreach(println(_));
  }

  def collectOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1, 2, 3, 4, 5));
    val array = dataRDD.collect();
    array.foreach(println(_));
  }

  def reduceOp(sc: SparkContext): Unit = {
    val dataRDD = sc.parallelize(Array(1, 2, 3, 4, 5));
    val sum = dataRDD.reduce(_ + _);
    println(sum)
  }

  private def getSparkContext = {
    //创建SparkContext
    val conf = new SparkConf();
    conf.setAppName("ActionOp").setMaster("local");
    new SparkContext(conf);
  }
}
